package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.EnvUtil;
import realtime.util.HbaseUtil;
import realtime.util.KafkaUtil;

public class DwdTradeOrderDetail {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        //env.setBufferTimeout(3000L);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_trade_order_detail_group"));
        Table orderDetail = tableEnv.sqlQuery("select" +
                " data['id'] id,\n" +
                " data['course_id'] course_id,\n" +
                " data['course_name'] course_name,\n" +
                " data['order_id'] order_id,\n" +
                " data['user_id'] user_id,\n" +
                " data['origin_amount'] origin_amount,\n" +
                " data['coupon_reduce'] coupon_reduce,\n" +
                " data['final_amount'] final_amount,\n" +
//                " data['session_id'] session_id,\n" +
                " data['create_time'] create_time,\n" +
                " data['update_time'] update_time," +
                " ts," +
//                " row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint)))," +
                " proc_time" +
                " from topic_db" +
                " where `table`='order_detail' and `type` = 'insert'");
        tableEnv.createTemporaryView("order_detail",orderDetail);

        Table orderInfo = tableEnv.sqlQuery("select \n" +
                " data['id'] id,\n" +
                " data['session_id'] session_id,\n" +
                " data['province_id'] province_id\n" +
                " from `topic_db`\n" +
                " where `table` = 'order_info'\n" +
                " and `type` = 'insert'");
        tableEnv.createTemporaryView("order_info", orderInfo);

        tableEnv.executeSql("create table source_log(" +
                " `page` map<string,string>," +
                " `common` map<string,string>," +
                " `actions` map<string,string>" +
                " ) " + KafkaUtil.getKafkaDDL("topic_log","dwd_trade_order_detail_log_group"));

        Table source = tableEnv.sqlQuery("select " +
                " common['sid'] sid," +
                " common['sc'] sc" +
                " from source_log ");
        tableEnv.createTemporaryView("source1", source);

        Table table = tableEnv.sqlQuery(" select sid,sc from source1 group by sid,sc");

        tableEnv.createTemporaryView("source",table);

        //tableEnv.executeSql("select * from source").print();

        Table result = tableEnv.sqlQuery("select" +
                " t1.id,\n" +
                " t1.course_id,\n" +
                " t1.course_name,\n" +
                " t1.order_id,\n" +
                " t1.user_id,\n" +
                " t1.origin_amount,\n" +
                " t1.coupon_reduce,\n" +
                " t1.final_amount,\n" +
                " t1.create_time,\n" +
                " t1.update_time," +
                " t1.ts," +
                " t2.session_id,\n" +
                " t2.province_id," +
                " t3.sc" +
                " from order_detail t1 join order_info t2" +
                " on t1.order_id=t2.id" +
                " join source t3 on t2.session_id=t3.sid" +
                " ");
        tableEnv.createTemporaryView("result",result);

        //tableEnv.executeSql("select * from `result`").print();

        tableEnv.executeSql("create table dwd_trade_order_detail(" +
                "  id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  order_id string,\n" +
                "  user_id string,\n" +
                "  origin_amount string,\n" +
                "  coupon_reduce string,\n" +
                "  final_amount string,\n" +
                "  create_time string,\n" +
                "  update_time string,\n" +
                "  ts string," +
                "  session_id string," +
                "  province_id string,\n" +
                "  sc string," +
                " primary key(id) not enforced" +
                " )" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_order_detail"));

        tableEnv.executeSql("insert into dwd_trade_order_detail select * from `result`");

    }
}
